package com.rpc.framework;

import com.rpc.framework.cdec.Decoder;
import com.rpc.framework.cdec.Encoder;
import com.rpc.framework.cdec.JsonDecoder;
import com.rpc.framework.cdec.JsonEncoder;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by tan on 2017/8/24.
 */
public class ServiceInvocationHandler implements InvocationHandler {

	private Socket socket;
	private Object target;
	private AtomicLong idGenerater;

	private Decoder decoder;
	private Encoder encoder;

	private Map<Long, ResponseFuture> requestIdResponseMap;

	private InputStream in;

	public ServiceInvocationHandler(String host, int port, Object target) throws IOException {
		this.socket = new Socket(host, port);
		this.target = target;
		this.decoder = new JsonDecoder();
		this.encoder = new JsonEncoder();
		this.requestIdResponseMap = new ConcurrentHashMap<>();
		this.in =  socket.getInputStream();
		this.idGenerater = new AtomicLong(0);

		Thread thread = new Thread(new SocketHandler());
		thread.start();
	}

	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		//创建Request对象
		Request request = new Request(idGenerater.getAndIncrement(),
				target.getClass().getName(), method.getName(), method.getParameterTypes(), args);

		OutputStream out = socket.getOutputStream();
		out.write(encoder.encode(request));
		//out.flush();
		//return method.invoke(target, args);

		ResponseFuture future = new ResponseFuture();
		requestIdResponseMap.put(request.getRequestId(),future);
		return future.get().getResult();
	}

	class SocketHandler implements Runnable {
		@Override
		public void run() {
			try {
				while (true) {
					byte[] bytes = new byte[1024];
					int size = in.read(bytes);
					if (size == 0) {
						continue;
					} else if (size == -1) {
						break;
					} else {
						ByteBuffer buffer = ByteBuffer.allocate(size);
						buffer.put(bytes, 0, size);
						byte[] tempBytes = new byte[size];
						buffer.flip();
						buffer.get(tempBytes);
						Response response = decoder.decode(tempBytes, Response.class);
						ResponseFuture future = requestIdResponseMap.get(response.getRequestId());
						requestIdResponseMap.remove(response.getRequestId());
						//ConcurrentHashMap的弱一致性，会导致获取到的future可能为null
						future.setResponse(response);
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
